package com.pig4cloud.pigx.contract.api.v1.websocket.subscribe;

import com.alibaba.fastjson.JSONObject;
import com.pig4cloud.pigx.common.core.util.R;
import com.pig4cloud.pigx.common.customize.api.enums.AbnormalEnum;
import com.pig4cloud.pigx.common.customize.api.exception.OpenApiException;
import com.pig4cloud.pigx.common.customize.i18n.util.MessageUtils;
import com.pig4cloud.pigx.contract.api.v1.utils.OpenApiContractUtilsV1;
import com.pig4cloud.pigx.contract.api.v1.websocket.dto.OpenApiSymbolDTOV1;
import com.pig4cloud.pigx.contract.api.v1.websocket.message.OpenApiSendMessageV1;
import com.pig4cloud.pigx.contract.api.v1.websocket.vo.OpenApiAbstractTopicVOV1;
import com.pig4cloud.pigx.contract.api.v1.websocket.vo.OpenApiSymbolLastPriceVOV1;
import com.pig4cloud.pigx.contract.mongo.entity.ContractSymbolMongo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 */
@Slf4j
@Component
public class OpenApiSymbolLastPriceV1 extends RequestDataAbstractTopicV1 {

    private final OpenApiContractUtilsV1 apiContractUtils;

    public static final String API_NAME = "symbol_last_price";

    public OpenApiSymbolLastPriceV1(OpenApiContractUtilsV1 apiContractUtils) {
        this.apiContractUtils = apiContractUtils;
        init();
    }

    private void init() {
        wsPush();
    }

    private void wsPush() {
        ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
        service.scheduleAtFixedRate(() -> {
            Map<String, List<WebSocketSession>> map = new HashMap<>();
            // 记录请求参数不合法的session,发送一次报错信息后,移除session队列
            List<WebSocketSession> requestErrorList = new ArrayList<>();
            // 根据参数对session分组
            for (Map.Entry<WebSocketSession, OpenApiAbstractTopicVOV1> entry : subList.entrySet()) {
                if (!entry.getKey().isOpen()) {
                    subList.remove(entry.getKey());
                    log.info("ws 移除已关闭的session");
                    continue;
                }
                try {
                    // 开始参数校验
                    JSONObject param = entry.getValue().getData().getRequestParam();
                    OpenApiSymbolDTOV1 dto = JSONObject.parseObject(param.toJSONString(), OpenApiSymbolDTOV1.class);
                    if (dto == null || CollectionUtils.isEmpty(dto.getSymbolList())) {
                        // 必填参数为空发送异常消息
                        entry.getKey().sendMessage(OpenApiSendMessageV1.getTextMessage(R.failed(
                                AbnormalEnum.MSG.MANDATORY_PARAM_EMPTY_OR_MALFORMED.getCode(),
                                MessageUtils.get(AbnormalEnum.MSG.MANDATORY_PARAM_EMPTY_OR_MALFORMED.getMessage()))
                        ));
                        requestErrorList.add(entry.getKey());
                        continue;
                    }

                    // 以币对名称为key对用户分组
                    dto.getSymbolList().forEach(symbolName -> map.computeIfAbsent(symbolName, key -> new ArrayList<>())
                            .add(entry.getKey()));

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            // 移除请求参数异常的session
            for (WebSocketSession session : requestErrorList) {
                unSub(session);
            }
            try {
                // 记录请求异常的session,发送一次报错信息后,移除session队列
                List<WebSocketSession> symbolErrorList = new ArrayList<>();
                map.forEach((key, value) -> {
                    // 币对校验
                    ContractSymbolMongo mongo;
                    try {
                        // 币对校验
                        mongo = apiContractUtils.getSymbolMongo(key);
                        OpenApiSymbolLastPriceVOV1 vo = new OpenApiSymbolLastPriceVOV1();
                        vo.setSymbol(mongo.getName());
                        vo.setPrice(mongo.getLastPrice());

                        for (WebSocketSession session : value) {
                            try {
                                // 发送币对最新价
                                session.sendMessage(OpenApiSendMessageV1.getTextMessage(
                                        R.ok(vo)
                                ));
                            } catch (IOException exception) {
                                throw new RuntimeException(exception);
                            }
                        }
                    } catch (OpenApiException e) {
                        // 返回异常消息
                        for (WebSocketSession session : value) {
                            try {
                                // 将捕获的异常消息发送
                                session.sendMessage(OpenApiSendMessageV1.getTextMessage(
                                        R.failed(e.getCode(), e.getMessage())
                                ));
                                symbolErrorList.add(session);
                            } catch (IOException exception) {
                                throw new RuntimeException(exception);
                            }
                        }
                    }
                });
                // 移除请求无效的session
                for (WebSocketSession session : symbolErrorList) {
                    unSub(session);
                }
            } catch (Exception e) {
                e.printStackTrace();
                log.error("webSocket---symbol_last_price---最新币对最新价格Error");
            }
        }, 0, 500, TimeUnit.MILLISECONDS);
    }

}
